package com.yunmaozj.tools.lock;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

import java.io.Serializable;
import java.security.AccessControlException;
import java.util.UUID;
import java.util.WeakHashMap;
import java.util.concurrent.*;

import static java.util.Objects.isNull;

/**
 * @author zhongminghong
 * @email zhongmh@yunmaozj.com
 * @create 2021/4/27
 */
public class RedisLock implements Serializable {
    private Logger logger = LoggerFactory.getLogger(RedisLock.class);
    private String host;
    private int port = 6379;

    private JedisPool jedisPool;
    private Executor executor = Executors.newSingleThreadExecutor();
    private final static String channel = "_redis_lock_channel";
    private final ThreadLocal<String> threadId = new ThreadLocal<>();
    private final ThreadLocal<String> businessIdThreadLocalMap = new ThreadLocal<>();
    private final ConcurrentHashMap<String, CyclicBarrier> cyclicBarrierConcurrentHashMap = new ConcurrentHashMap<>();
    private final WeakHashMap<String, Thread> threadWeakHashMap = new WeakHashMap<>();
    private final Lock lock;

    public RedisLock(String host, int port, LockType lockType) {
        this.host = host;
        this.port = port;
        if (lockType == LockType.DEFAULT) {
            lock = new DefaultLock();
        } else {
            lock = new ValveLock();
        }
    }

    public RedisLock(String host, LockType lockType) {
        this(host, 6379, lockType);
    }

    public void afterProperties() {
        GenericObjectPoolConfig config = new GenericObjectPoolConfig();
        config.setMaxTotal(100);
        this.jedisPool = new JedisPool(config,this.host, this.port);
        executor.execute(() -> jedisPool.getResource().subscribe(new JedisPubSub() {
            @Override
            public void onMessage(String channel, String businessId) {
                logger.debug("receive channel[" + channel + "] message[" + businessId + "]");
                if (cyclicBarrierConcurrentHashMap.containsKey(businessId)) {
                    CyclicBarrier cyclicBarrier = getCyclicBarrier(businessId);
                    cyclicBarrier.reset();
                }
            }
        }, channel));
    }

    public void lock(String businessId) {
        lock.lock(businessId);
        businessIdThreadLocalMap.set(businessId);
    }

    public boolean tryLock(String businessId) {
        Jedis jedis = jedisPool.getResource();
        boolean result = lock.tryLock(jedis, businessId);
        jedis.close();
        return result;
    }

    public boolean tryLock(String businessId, long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    public void unlock() {
        unlock(businessIdThreadLocalMap.get());

    }

    public void unlock(String businessId) {
        Jedis jedis = jedisPool.getResource();
        lock.unlock(jedis, businessId);
        jedis.close();
    }

    private CyclicBarrier getCyclicBarrier(String businessId) {
        if (!cyclicBarrierConcurrentHashMap.containsKey(businessId)) {
            synchronized (cyclicBarrierConcurrentHashMap) {
                if (!cyclicBarrierConcurrentHashMap.containsKey(businessId)) {
                    cyclicBarrierConcurrentHashMap.put(businessId, new CyclicBarrier(Integer.MAX_VALUE));
                }
            }
        }
        return cyclicBarrierConcurrentHashMap.get(businessId);
    }

    interface Lock {
        boolean tryLock(Jedis jedis, String businessId);

        boolean tryLock(Jedis jedis, String businessId, long time, TimeUnit unit);

        void lock(String businessId);

        void unlock(Jedis jedis, String businessId);
    }

    class DefaultLock implements Lock {

        public void lock(String businessId) {
            boolean isDone = false;
            String id = UUID.randomUUID().toString();
            threadId.set(id);
            CyclicBarrier cyclicBarrier = getCyclicBarrier(businessId);
            do {
                Jedis jedis = jedisPool.getResource();
                try {
                    isDone = compareAndSetState(jedis, businessId);
                    if (!isDone) {
                        cyclicBarrier.await(1, TimeUnit.SECONDS);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    isDone = compareAndSetState(jedis, businessId);
                } catch (TimeoutException e) {
                    isDone = compareAndSetState(jedis, businessId);
                }
                jedis.close();
            } while (!isDone);
        }

        private boolean compareAndSetState(Jedis jedis, String businessId) {
            boolean result = jedis.setnx(businessId, threadId.get()) == 1;
            return result;
        }

        public boolean tryLock(Jedis jedis, String businessId) {
            return false;
        }

        public boolean tryLock(Jedis jedis, String businessId, long time, TimeUnit unit) {
            return false;
        }

        public void unlock(Jedis jedis, String businessId) {
            String id = threadId.get();

            jedis.del(businessId);
            logger.debug("delete businessId {}", businessId);
            jedis.publish(channel, businessId);
            logger.debug("publish businessId {}", businessId);
            threadWeakHashMap.remove(id);
            threadId.remove();
        }
    }

    class ValveLock implements Lock {

        private int defaultMaxHolderNumber = 10;

        @Override
        public boolean tryLock(Jedis jedis, String businessId) {
            return jedis.exists(businessId) && jedis.incr(businessId) < defaultMaxHolderNumber
                    ? true : jedis.setnx(businessId, "1") == 1 ? true : jedis.incr(businessId) < defaultMaxHolderNumber;
        }

        @Override
        public boolean tryLock(Jedis jedis, String businessId, long time, TimeUnit unit) {
            return false;
        }

        @Override
        public void lock(String businessId) {
            boolean isDone;
            {
                Jedis jedis = jedisPool.getResource();
                isDone = tryLock(jedis, businessId);
                if (isDone) {
                    threadId.set(businessId);
                    jedis.close();
                } else {
                    CyclicBarrier cyclicBarrier = getCyclicBarrier(businessId);
                    try {
                        cyclicBarrier.await(1, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        //e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        //e.printStackTrace();
                    } catch (TimeoutException e) {
                        //e.printStackTrace();
                    }
                }
            }
            while (isDone) ;
        }

        @Override
        public void unlock(Jedis jedis, String businessId) {
            jedis.publish(channel, businessId);
            jedis.decr(businessId);
            threadId.remove();
        }
    }

    public enum LockType {
        DEFAULT, VALVE;
    }

}
